package cn.wx.read.kafka.pool;

import java.io.BufferedInputStream;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.OutputStreamWriter;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.csvreader.CsvReader;

public class ExecutorPool {
	
	Logger logger = LoggerFactory.getLogger(ExecutorPool.class);
	
	/**
	 * 定义线程数
	 */
	public final static Integer tc = 6;
	
	/**
	 * 线程池
	 */
	private ExecutorService threadPool;
	
	/**
	 * 初始化线程
	 */
	public void initPool() {
		threadPool = Executors.newFixedThreadPool(tc);
	}
	
	/**
	 * 向线程池中加入线程
	 * @param r
	 */
	public void execute(Runnable r) {
		threadPool.execute(r);
	}
	
	/**
	 * 判断当前线程池中是否还有线程执行
	 */
	public Boolean isTerminating() {
		if(threadPool.isTerminated()) {
			return true;
		}
		return false;
	}
	
	/**
	 * 判断当前线程池中是否还有线程执行
	 */
	public void poolShutdown() {
		threadPool.shutdown();
	}
	
	/**
	 * 锁定当前线程
	 */
	
	public void lock() {
		
		long s_ = System.currentTimeMillis();
		
		while(true){
    		if(threadPool.isTerminated()) {
    			long e_ = System.currentTimeMillis();
    			logger.info("执行时长:" + (float)(e_-s_)/1000);
				break;
			}
    		logger.info("心跳监控，线程执行中！");
			try {
				Thread.sleep(10000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}  
		}
	}
	
	
	/**
	 * 读取csv
	 */
	public CsvReader getCsvReader(String file_) throws FileNotFoundException {
		File file = new File(file_);
        FileInputStream fileInputStream = new FileInputStream(file);  
        BufferedInputStream bufferedInputStream = new BufferedInputStream(fileInputStream);
		return new CsvReader(bufferedInputStream,Charset.forName("utf-8"));
	}
	
	public CsvReader getCsvReader(String filePath,String fileName) throws FileNotFoundException {
		return getCsvReader(filePath + fileName);
	}
	
	/**
	 * 分片csv，返回分片个数
	 */
	public Integer zoneCsv(String filePath,String fileName,String zoneName){
		Integer zc = 0;
		Integer result = 0;
		try {
			CsvReader csvReader = getCsvReader(filePath,fileName);
			StringBuffer sb = new StringBuffer();;
			BufferedWriter output = null;
			// 读表头
			csvReader.readHeaders();
			String[] head = csvReader.getHeaders();
			
			for(int hi=0;hi<head.length;hi++) {
				if(hi == head.length -1) {
					sb.append(head[hi]);
					break;
				}
				sb.append(head[hi]).append(",");
			}
			
            int i = 0;
            while (csvReader.readRecord()){
            	
            	if(i == 0) {
            		result++;
            		File of = new File(filePath + zoneName + zc + ".csv");
        			FileOutputStream ofs = new FileOutputStream(of,false);
        			OutputStreamWriter ofw = new OutputStreamWriter(ofs,"utf-8");
        			output = new BufferedWriter(ofw);
					output.write(sb.toString());
					output.newLine();
            	}
            	
            	
                // 读一整行
            	String line = csvReader.getRawRecord();
				System.out.println("读取" + (i++) +"行");
				output.write(line);
				output.newLine();
				output.flush();
				if(i > 20000) {
					i = 0;
					zc++;
				}
            }
            output.flush();
            output.close();
            output = null;
            
            /**
			 * 关闭读入流
			 */
			csvReader.close();
			csvReader = null;

		} catch (Exception e) {
			e.printStackTrace();
		}
		
		return result;
		
	}
	
}
